
package wal

import (
	"bytes"
	"errors"
	"fmt"
	"go.etcd.io/etcd/raft"
	"hash/crc32"
	"io"
	"os"
	"path/filepath"
	"strings"
	"sync"
	"time"

	"go.etcd.io/etcd/client/pkg/v3/fileutil"
	"go.etcd.io/etcd/pkg/v3/pbutil"
	"go.etcd.io/etcd/raft/v3"
	"go.etcd.io/etcd/raft/v3/raftpb"
	"go.etcd.io/etcd/server/v3/storage/wal/walpb"

	"go.uber.org/zap"
)

const (
	metadataType int64 = iota + 1
	entryType
	stateType
	crcType
	snapshotType

	// warnSyncDuration是在
	// 记录警告之前分配给fsync的时间量
	warnSyncDuration = time.Second
)

var (
	// SegmentSizeBytes是每个wal段文件的预分配大小。
	// 实际大小可能大于此值。一般情况下，应该使用默认的
	// 值，但它被定义为导出变量
	// 以便测试可以设置不同的段大小。
	SegmentSizeBytes int64 = 64 * 1000 * 1000 // 64MB 

	ErrMetadataConflict             = errors.New("wal: conflicting metadata found")
	ErrFileNotFound                 = errors.New("wal: file not found")
	ErrCRCMismatch                  = errors.New("wal: crc mismatch")
	ErrSnapshotMismatch             = errors.New("wal: snapshot mismatch")
	ErrSnapshotNotFound             = errors.New("wal: snapshot not found")
	ErrSliceOutOfRange              = errors.New("wal: slice bounds out of range")
	ErrMaxWALEntrySizeLimitExceeded = errors.New("wal: max entry size limit exceeded")
	ErrDecoderNotFound              = errors.New("wal: decoder not found")
	crcTable                        = crc32.MakeTable(crc32.Castagnoli)
)

// 结构体WAL，它对外提供了WAL日志文件管理的核心API。在操纵WAL日志时，
// 对应的WAL实例有read和append两种模式，
// 新创建的WAL实例处于append模式，该模式下只能向WAL中追加日志。
// 当恢复一个节点时（例如，者机节点的重启〉，就需要读取WAL日志的内容，
// 此时刚打开的WAL实例处于read模式，它只能读取日志记录，当读取完全部的日志之后，
// WAL实例转换成append模式，可以继续向其追加日志记录。
type WAL struct {
	lg *zap.Logger
	dir string // 存放WAL日志文件的目录路径。
	dirFile *os.File	// 根据dir路径创建的File实例。
	metadata []byte           // 在每个WAL日志文件的头部，都会写入metadata元数据。
	// WAL日志记录的追加是批量的，在每次批量写入entry Type类型的日志之后，
	// 都会再追加一条stateType类型的日志记录，在HardState中记录了当前的Term、
	// 当前节点的投票结果和己提交日志的位置。
	state    raftpb.HardState
	// 每次读取WAL日志时，并不会每次都从头开始读取，而是通过这里的start宇段指定具体的起始位置。
	// walpb.Snapshot中的Index字段记录了对应快照数据所涵盖的最后一条Entry记录的索引值，
	// Term字段则记录了对应Entry记录的Term值。在读取WAL日志文件时，
	// 我们就可以根据这些信息，找到合适的位置并开始读取记录。
	start     walpb.Snapshot
	// 负责在读取WAL日志文件时，将二进制数据反序列化成Record实例。
	decoder   *decoder
	readClose func() error   // closer for decode reader
	unsafeNoSync bool // 如果设置，请不要同步
	// 读写WAL日志时需要加锁同步。
	mu      sync.Mutex
	// WAL中最后一条Entry记录的索引值。
	enti    uint64
	// 负责将写入WAL日志文件的Record实例进行序列化成二进制数据。
	encoder *encoder
	// 当前WAL实例管理的所有WAL日志文件对应的句柄。
	locks []*fileutil.LockedFile
	// filePipeline实例负责创建新的临时文件。
	fp    *filePipeline
}

// wal.Create（）方法，该方法不仅会创建WAL实例，而是做了很多初始化工作，其大致步骤如下：
// ( 1 ）创建临时目录，并在临时目录中创建编号为“0－0”的WAL日志文件，
// WAL日志文件名由两部分组成，一部分是seq（单调递增），另一部分是该日志文件中的第一条日志记录的索引值。
// (2）尝试为该WAL日志文件预分配磁盘空间。
// (3）向该WAL日志文件中写入一条crcType类型的日志记录、一条metadataType类型的日志记录及一条snapshotType类型的日志记录。
// (4）创建WAL实例关联的filePipeline实例。
// (5）将临时目录重命名为WAL.dir字段指定的名称。
// 之所以先使用临时目录完成初始化操作再将其重命名的方式，主要是为了让整个初始化过程看上去是一个原子操作。
func Create(lg *zap.Logger, dirpath string, metadata []byte) (*WAL, error) {
	// 检测文件夹是否存在
	if Exist(dirpath) {
		return nil, os.ErrExist
	}

	if lg == nil {
		lg = zap.NewNop()
	}

	tmpdirpath := filepath.Clean(dirpath) + ".tmp"	// 得到临时目录的路径
	if fileutil.Exist(tmpdirpath) {	// 清空临时目录中的文件
		if err := os.RemoveAll(tmpdirpath); err != nil {
			return nil, err
		}
	}
	defer os.RemoveAll(tmpdirpath)

	if err := fileutil.CreateDirAll(lg, tmpdirpath); err != nil {	// 创建临时文件夹
		lg.Warn(
			"failed to create a temporary WAL directory",
			zap.String("tmp-dir-path", tmpdirpath),
			zap.String("dir-path", dirpath),
			zap.Error(err),
		)
		return nil, err
	}
	// 	第一个WAL日志文件的路径（文件名为0-0)
	p := filepath.Join(tmpdirpath, walName(0, 0))
	// 创建临时文件，注意文件的模式和权限，
	f, err := fileutil.LockFile(p, os.O_WRONLY|os.O_CREATE, fileutil.PrivateFileMode)
	if err != nil {
		lg.Warn(
			"failed to flock an initial WAL file",
			zap.String("path", p),
			zap.Error(err),
		)
		return nil, err
	}
	// 移动临时文件的offset到文件结尾处，
	// 注意Seek（）方法的第二个参数（0是相对文件开头，1是相对当前offset,2是相对文件结尾）
	if _, err = f.Seek(0, io.SeekEnd); err != nil {
		lg.Warn(
			"failed to seek an initial WAL file",
			zap.String("path", p),
			zap.Error(err),
		)
		return nil, err
	}
	// 对新建的临时文件进行空间预分配，默认值是64MB(SegmentSizeBytes)
	if err = fileutil.Preallocate(f.File, SegmentSizeBytes, true); err != nil {
		lg.Warn(
			"failed to preallocate an initial WAL file",
			zap.String("path", p),
			zap.Int64("segment-bytes", SegmentSizeBytes),
			zap.Error(err),
		)
		return nil, err
	}

	w := &WAL{	// 创建WAL实例
		lg:       lg,
		dir:      dirpath,	// 存放WAL日志文件的目录的路径
		metadata: metadata,	// 元数据
	}
	// 创建写WAL日志文件的encoder，
	w.encoder, err = newFileEncoder(f.File, 0)
	if err != nil {
		return nil, err
	}
	// 将WAL日志文件对应的LockedFile实例记录到locks字段中，表示当前WAL实例正在管理该日志文件
	w.locks = append(w.locks, f)
	// 创建一条ereType类型的日志写入WAL日志文件，
	if err = w.saveCrc(0); err != nil {
		return nil, err
	}
	// 将元数据封装成一条metadataType类型的日志记录写入WAL日志文件
	if err = w.encoder.encode(&walpb.Record{Type: metadataType, Data: metadata}); err != nil {
		return nil, err
	}
	// 创建一条空的snapshotType类型的日志记录写入临时文件，
	if err = w.SaveSnapshot(walpb.Snapshot{}); err != nil {
		return nil, err
	}

	logDirPath := w.dir
	// 将临时目录重命名，并创建WAL实例关联的filePipline实例，
	if w, err = w.renameWAL(tmpdirpath); err != nil {
		lg.Warn(
			"failed to rename the temporary WAL directory",
			zap.String("tmp-dir-path", tmpdirpath),
			zap.String("dir-path", logDirPath),
			zap.Error(err),
		)
		return nil, err
	}

	var perr error
	defer func() {
		if perr != nil {
			w.cleanupWAL(lg)
		}
	}()

	// 临时目录重命名之后，需妥将重命名操作刷新到磁盘上，
	pdir, perr := fileutil.OpenDir(filepath.Dir(w.dir))
	if perr != nil {
		lg.Warn(
			"failed to open the parent data directory",
			zap.String("parent-dir-path", filepath.Dir(w.dir)),
			zap.String("dir-path", w.dir),
			zap.Error(perr),
		)
		return nil, perr
	}
	dirCloser := func() error {
		if perr = pdir.Close(); perr != nil {
			lg.Warn(
				"failed to close the parent data directory file",
				zap.String("parent-dir-path", filepath.Dir(w.dir)),
				zap.String("dir-path", w.dir),
				zap.Error(perr),
			)
			return perr
		}
		return nil
	}
	start := time.Now()
	// 同步磁盘的操作，
	if perr = fileutil.Fsync(pdir); perr != nil {
		dirCloser()
		lg.Warn(
			"failed to fsync the parent data directory file",
			zap.String("parent-dir-path", filepath.Dir(w.dir)),
			zap.String("dir-path", w.dir),
			zap.Error(perr),
		)
		return nil, perr
	}
	walFsyncSec.Observe(time.Since(start).Seconds())
	if err = dirCloser(); err != nil {
		return nil, err
	}

	return w, nil	// 返回WAL实例
}

func (w *WAL) Reopen(lg *zap.Logger, snap walpb.Snapshot) (*WAL, error) {
	err := w.Close()
	if err != nil {
		lg.Panic("failed to close WAL during reopen", zap.Error(err))
	}
	return Open(lg, w.dir, snap)
}

func (w *WAL) SetUnsafeNoFsync() {
	w.unsafeNoSync = true
}

func (w *WAL) cleanupWAL(lg *zap.Logger) {
	var err error
	if err = w.Close(); err != nil {
		lg.Panic("failed to close WAL during cleanup", zap.Error(err))
	}
	brokenDirName := fmt.Sprintf("%s.broken.%v", w.dir, time.Now().Format("20060102.150405.999999"))
	if err = os.Rename(w.dir, brokenDirName); err != nil {
		lg.Panic(
			"failed to rename WAL during cleanup",
			zap.Error(err),
			zap.String("source-path", w.dir),
			zap.String("rename-path", brokenDirName),
		)
	}
}

// WAL.rename Wal（）方法在不同系统平台的操作略有不同，
// 但是主要操作都是重命名临时目录并创建关联的filePipeline实例。
func (w *WAL) renameWAL(tmpdirpath string) (*WAL, error) {
	if err := os.RemoveAll(w.dir); err != nil {	// 清空wal文件夹
		return nil, err
	}
	if err := os.Rename(tmpdirpath, w.dir); err != nil {	// 重命名临时文件夹
		if _, ok := err.(*os.LinkError); ok {
			return w.renameWALUnlock(tmpdirpath)
		}
		return nil, err
	}
	// 创建WAL实例关联的filePipeline实例)
	w.fp = newFilePipeline(w.lg, w.dir, SegmentSizeBytes)
	df, err := fileutil.OpenDir(w.dir)
	w.dirFile = df	// WAL.dirFile字段记录了WAL日志目录对应的文件句柄
	return w, err
}

func (w *WAL) renameWALUnlock(tmpdirpath string) (*WAL, error) {
	// 带锁定文件的目录重命名在windows/cifs上不起作用；
	// 关闭WAL以释放锁，以便可以重命名目录。
	w.lg.Info(
		"closing WAL to release flock and retry directory renaming",
		zap.String("from", tmpdirpath),
		zap.String("to", w.dir),
	)
	w.Close()

	if err := os.Rename(tmpdirpath, w.dir); err != nil {
		return nil, err
	}

	// 重新打开并重新锁定
	newWAL, oerr := Open(w.lg, w.dir, walpb.Snapshot{})
	if oerr != nil {
		return nil, oerr
	}
	if _, _, _, err := newWAL.ReadAll(); err != nil {
		newWAL.Close()
		return nil, err
	}
	return newWAL, nil
}

// Open（）函数创建的WAL实例读取完全部日志后，可以继续追加日志：
func Open(lg *zap.Logger, dirpath string, snap walpb.Snapshot) (*WAL, error) {
	w, err := openAtIndex(lg, dirpath, snap, true)
	if err != nil {
		return nil, err
	}
	if w.dirFile, err = fileutil.OpenDir(w.dir); err != nil {
		return nil, err
	}
	return w, nil
}

// OpenForRead（）函数创建的WAL实例只能用于读取日志，不能追加日志。
func OpenForRead(lg *zap.Logger, dirpath string, snap walpb.Snapshot) (*WAL, error) {
	return openAtIndex(lg, dirpath, snap, false)
}

// 需要注意snap和write参数，snap.Index指定了日志读取的起始位置
// write参数则指定了打开日志文件的模式
func openAtIndex(lg *zap.Logger, dirpath string, snap walpb.Snapshot, write bool) (*WAL, error) {
	if lg == nil {
		lg = zap.NewNop()
	}
	names, nameIndex, err := selectWALFiles(lg, dirpath, snap)
	if err != nil {
		return nil, err
	}

	rs, ls, closer, err := openWALFiles(lg, dirpath, names, nameIndex, write)
	if err != nil {
		return nil, err
	}

	// 创建WAL实例
	w := &WAL{
		lg:        lg,
		dir:       dirpath,
		start:     snap,	// 记录Snapshot的信息
		// 创建用于读取日志记录的decoder实例，这里并没有初始化encoder，所以还不能写入日志记录
		decoder:   newDecoder(rs...),
		// 如果是只读模式，在读取完全部日志文件之后，则会调用该方法关闭所有日志文件
		readClose: closer,
		// 当前WAL实例管理的日志文件
		locks:     ls,
	}

	if write {
		// 如果是读写模式，读取完全部日志文件之后，由于后续有追加操作，所以不需妥关l羽日志文件；
		// 另外，还要为WAL实例创建关驳的filePipeline实例，用于产生新的日志文件
		w.readClose = nil
		if _, _, err := parseWALName(filepath.Base(w.tail().Name())); err != nil {
			closer()
			return nil, err
		}
		w.fp = newFilePipeline(lg, w.dir, SegmentSizeBytes)	// 创建filePipeline
	}

	return w, nil
}

// 获取全部的WAL日志文件名，并且这些文件名会进行排序，
func selectWALFiles(lg *zap.Logger, dirpath string, snap walpb.Snapshot) ([]string, int, error) {
	// 获取全部的WAL日志文件名，并且这些文件名会进行排序，
	names, err := readWALNames(lg, dirpath)
	if err != nil {
		return nil, -1, err
	}
	// 根据WAL日志文件名的规则，查找上面得到的所有文件名，找到index最大且index小于snap.Index的WAL日志文件，\
	// 并返回该文件在names数组中的索引（nameIndex) ,
	nameIndex, ok := searchIndex(lg, names, snap.Index)
	if !ok || !isValidSeq(lg, names[nameIndex:]) {
		err = ErrFileNotFound
		return nil, -1, err
	}

	return names, nameIndex, nil
}

func openWALFiles(lg *zap.Logger, dirpath string, names []string, nameIndex int, write bool) ([]io.Reader, []*fileutil.LockedFile, func() error, error) {
	rcs := make([]io.ReadCloser, 0)
	rs := make([]io.Reader, 0)	// 注意该切片中元素的类型，在decoder中会使用到
	ls := make([]*fileutil.LockedFile, 0)
	// 从nameIndex开始读取剩余的WAL日志文件
	for _, name := range names[nameIndex:] {
		p := filepath.Join(dirpath, name)	// 获取WAL日志的绝对路径
		if write {	// 以读写模式打开WAL日志文件
			// 打开WAL日志文件并且对文件加锁，注意打开文件的模式，
			l, err := fileutil.TryLockFile(p, os.O_RDWR, fileutil.PrivateFileMode)
			if err != nil {
				closeAll(lg, rcs...)
				return nil, nil, nil, err
			}
			// 文件句柄记录到ls和res这两个切片中
			ls = append(ls, l)
			rcs = append(rcs, l)
		} else {
			// 如果write参数为false，则表示以只读模式打开文件，
			rf, err := os.OpenFile(p, os.O_RDONLY, fileutil.PrivateFileMode)
			if err != nil {
				closeAll(lg, rcs...)
				return nil, nil, nil, err
			}
			// 注意，这里只将文件句柄添加到了res切片中
			ls = append(ls, nil)
			rcs = append(rcs, rf)
		}
		// 将文件句柄记录到rs切片中
		rs = append(rs, rcs[len(rcs)-1])
	}

	closer := func() error { return closeAll(lg, rcs...) }	// 后面关闲文件时，会调用该函数

	return rs, ls, closer, nil
}

// WAL.ReadAll（）方法首先从WAL.start字段指定的位置开始读取日志记录，
// 读取完毕之后，会根据读取的情况进行一系列异常处理。
// 然后根据当前WAL实例的模式进行不同的处理：
// 如果处于读写模式，
// 则需要先对后续的WAL日志文件进行填充并初始化WAL.encoder字段，为后面写入日志做准备；
// 如果处于只读模式下，
// 则需要关闭所有的日志文件。
// 另外需要注意的是，WAL.ReadAll（）方法的几个返回值都是从日志记录中读取到的。
func (w *WAL) ReadAll() (metadata []byte, state raftpb.HardState, ents []raftpb.Entry, err error) {
	w.mu.Lock()
	defer w.mu.Unlock()

	rec := &walpb.Record{}	// 创建Record实例

	if w.decoder == nil {
		return nil, state, nil, ErrDecoderNotFound
	}
	// 解码器，负责读取日志文件，并将日志数据反序列化成Record实例
	decoder := w.decoder

	var match bool	// 标识是否找到了start字段对应的日志记录
	// 循环读取WAL日志文件中的数据，多个WAL日志文件的切才是是在decoder中完成的，
	for err = decoder.decode(rec); err == nil; err = decoder.decode(rec) {
		switch rec.Type {	// 根据日志记录的类型进行分类处理
		case entryType:	// 读取到entryType类型的日志
			e := mustUnmarshalEntry(rec.Data)	// 反序列化Record.Data中记录的数据，得到Entry实例
			// /0<=e.Index-w.start。索引-1<len（ents）
			if e.Index > w.start.Index {	// 将start之后的Entry记录添加到ents中保存
				// /prevent“panic:runtime error:slice界限超出范围[：13038096702221461992]，容量为0”
				up := e.Index - w.start.Index - 1
				if up > uint64(len(ents)) {
					// /append调用前返回错误导致运行时恐慌
					return nil, state, nil, ErrSliceOutOfRange
				}
				// /下面的行可能会覆盖一些“未提交”项。
				ents = append(ents[:up], e)
			}
			w.enti = e.Index	// 记录读取到的最后一条Entry记录的索引值

		case stateType:	// 读取到stateType类型的日志记录
			state = mustUnmarshalState(rec.Data)	// 更新待返回的HardState状态信息

		case metadataType:	// 读取到metadataType类型的日志记录
			// 检测metadata数据是否发生冲突，如果冲突，则抛出异常
			if metadata != nil && !bytes.Equal(metadata, rec.Data) {
				state.Reset()
				return nil, state, nil, ErrMetadataConflict
			}
			metadata = rec.Data	// 更新待返回的元数据

		case crcType:	// 读取到ereType类型的日志记录，更新decoder中的ere信息
			crc := decoder.crc.Sum32()
			if crc != 0 && rec.Validate(crc) != nil {
				state.Reset()
				return nil, state, nil, ErrCRCMismatch
			}
			decoder.updateCRC(rec.Crc)	// 更新deeodr.crc字段，

		case snapshotType:	// 读取到snapshotType类型的日志记录
			var snap walpb.Snapshot
			pbutil.MustUnmarshal(&snap, rec.Data)	// 解析快照相关的数据
			if snap.Index == w.start.Index {
				if snap.Term != w.start.Term {
					state.Reset()
					return nil, state, nil, ErrSnapshotMismatch
				}
				match = true
			}

		default:	// 其他未知类型的日志记录，返回异常
			state.Reset()
			return nil, state, nil, fmt.Errorf("unexpected block type %d", rec.Type)
		}
	}
	// 到这里，读取WAL日志文件的操作就完成了（当然，中间可能出现了异常）

	switch w.tail() {	// 根据WAL.locks字段是否有位判断当前WAL是什么模式
	case nil:
		// 对于只读模式，并不需要将全部的日志都读出来，因为以只读模式打开WAL日志文件时，
		// 并没有加锁，所以最后一条日志记录可能只写了一半，从而导致io.ErrUnexpectedEOF异常
		if err != io.EOF && err != io.ErrUnexpectedEOF {
			state.Reset()
			return nil, state, nil, err
		}
	default:
		// 对于读写模式，则需要将日志记录全部读出来，所以此处不是EOF异常，则报错
		if err != io.EOF {
			state.Reset()
			return nil, state, nil, err
		}
		// 将文件指针移动到读取结束的位置，并将文件后续部分全部填充为。
		if _, err = w.tail().Seek(w.decoder.lastOffset(), io.SeekStart); err != nil {
			return nil, state, nil, err
		}
		if err = fileutil.ZeroToEnd(w.tail().File); err != nil {
			return nil, state, nil, err
		}
	}

	err = nil
	if !match {	// 如采在读取过程中没有找到与start对应的日志记录，则抛出异常
		err = ErrSnapshotNotFound
	}

	// 如采是只读模式，则关闭所有日志文件
	if w.readClose != nil {
		w.readClose()	// WAL.readClose实际指向的是WAL.CloseAll （）方法
		w.readClose = nil
	}
	w.start = walpb.Snapshot{}	// ／清空start字段

	w.metadata = metadata

	if w.tail() != nil {	// 如采是读写模式，则初始化WAL.encoder字段，为后面写入日志做准备
		w.encoder, err = newFileEncoder(w.tail().File, w.decoder.lastCRC())
		if err != nil {
			return
		}
	}
	w.decoder = nil	// 清空WAL.decoder字段，后续不能再用该WAL实例进行读取了

	return metadata, state, ents, err
}

// ValidSnapshotEntries返回给定目录中wal日志中的所有有效快照项。
// 如果快照项的索引小于或等于最近提交的硬状态，则快照项有效。
func ValidSnapshotEntries(lg *zap.Logger, walDir string) ([]walpb.Snapshot, error) {
	var snaps []walpb.Snapshot
	var state raftpb.HardState
	var err error

	rec := &walpb.Record{}
	names, err := readWALNames(lg, walDir)
	if err != nil {
		return nil, err
	}

	// 在读取模式下打开wal文件，以便在写入模式下在其他位置打开同一wal时不会发生冲突
	// 从wal文件上的读卡器创建新解码器
	rs, _, closer, err := openWALFiles(lg, walDir, names, 0, false)
	if err != nil {
		return nil, err
	}
	defer func() {
		if closer != nil {
			closer()
		}
	}()

	decoder := newDecoder(rs...)

	for err = decoder.decode(rec); err == nil; err = decoder.decode(rec) {
		switch rec.Type {
		case snapshotType:
			var loadedSnap walpb.Snapshot
			pbutil.MustUnmarshal(&loadedSnap, rec.Data)
			snaps = append(snaps, loadedSnap)
		case stateType:
			state = mustUnmarshalState(rec.Data)
		case crcType:
			crc := decoder.crc.Sum32()
			// 解码器的当前crc必须与记录的crc匹配。
			// 不需要匹配0 crc，因为在这种情况下解码器是新的。
			if crc != 0 && rec.Validate(crc) != nil {
				return nil, ErrCRCMismatch
			}
			decoder.updateCRC(rec.Crc)
		}
	}
	// 我们不必读取所有WAL条目
	// 因为解码器是在读取模式下打开的。
	if err != io.EOF && err != io.ErrUnexpectedEOF {
		return nil, err
	}

	// 筛选出比提交的硬状态更新的所有快照
	n := 0
	for _, s := range snaps {
		if s.Index <= state.Commit {
			snaps[n] = s
			n++
		}
	}
	snaps = snaps[:n:n]
	return snaps, nil
}

// 验证读取给定WAL并验证其未损坏。
// 它创建一个新的解码器来读取给定WAL的记录。
// 与任何打开的WAL都不冲突，但建议不要在打开WAL写入后调用此函数。
// 如果无法读取所需的快照，将返回ErrSnapshotNotFound。
// 如果加载的快照与预期的快照不匹配，它将返回错误errsnapshotmatch。
func Verify(lg *zap.Logger, walDir string, snap walpb.Snapshot) (*raftpb.HardState, error) {
	var metadata []byte
	var err error
	var match bool
	var state raftpb.HardState

	rec := &walpb.Record{}

	if lg == nil {
		lg = zap.NewNop()
	}
	names, nameIndex, err := selectWALFiles(lg, walDir, snap)
	if err != nil {
		return nil, err
	}

	// 以读取模式打开wal文件，这样就不会有冲突
	// 当以写入模式在其他位置打开同一wal时
	rs, _, closer, err := openWALFiles(lg, walDir, names, nameIndex, false)
	if err != nil {
		return nil, err
	}
	defer func() {
		if closer != nil {
			closer()
		}
	}()

	// 从wal文件上的读卡器创建新的解码器
	decoder := newDecoder(rs...)

	for err = decoder.decode(rec); err == nil; err = decoder.decode(rec) {
		switch rec.Type {
		case metadataType:
			if metadata != nil && !bytes.Equal(metadata, rec.Data) {
				return nil, ErrMetadataConflict
			}
			metadata = rec.Data
		case crcType:
			crc := decoder.crc.Sum32()
			// 解码器的当前crc必须与记录的crc匹配。
			// 我们不需要匹配0 crc，因为此时解码器是新的。
			if crc != 0 && rec.Validate(crc) != nil {
				return nil, ErrCRCMismatch
			}
			decoder.updateCRC(rec.Crc)
		case snapshotType:
			var loadedSnap walpb.Snapshot
			pbutil.MustUnmarshal(&loadedSnap, rec.Data)
			if loadedSnap.Index == snap.Index {
				if loadedSnap.Term != snap.Term {
					return nil, ErrSnapshotMismatch
				}
				match = true
			}
		// 我们忽略所有条目和状态类型记录，因为这些
		// 对于验证WAL内容不是必需的
		case entryType:
		case stateType:
			pbutil.MustUnmarshal(&state, rec.Data)
		default:
			return nil, fmt.Errorf("unexpected block type %d", rec.Type)
		}
	}

	// 我们不必读取所有WAL条目
	// 因为解码器是以读取模式打开的。
	if err != io.EOF && err != io.ErrUnexpectedEOF {
		return nil, err
	}

	if !match {
		return nil, ErrSnapshotNotFound
	}

	return &state, nil
}

// cut关闭当前写入的文件，并创建一个新文件，准备追加。
// cut首先创建一个临时wal文件，并将必要的头写入其中。
// 然后按原子方式将临时wal文件重命名为wal文件。
//
// WAL.cut（）方法首先通过filePipeline获取一个新建的临时文件，
// 然后写入ereType类型、metaType类型、stateType类型等必要日志记录（这个步骤与Create（）方法类似），
// 然后将临时文件重命名成符合WAL日志命名规范的新日志文件，
// 并创建对应的encoder实例更新到WAL.encoder字段
func (w *WAL) cut() error {
	// 获取当前日志文件的文件指针位置，
	off, serr := w.tail().Seek(0, io.SeekCurrent)
	if serr != nil {
		return serr
	}
	// 根据当前的文件指针位置，将后续填充内容Truncate掉，这主要是处理提早切换和
	// 预分配空间未使用的情况，truncate后可以释放该日志文件后续未使用的空间，
	if err := w.tail().Truncate(off); err != nil {
		return err
	}
	// 紧接着执行一次WAL.sync（）方法，将修改同步刷新到磁盘上，
	if err := w.sync(); err != nil {
		return err
	}
	// 根据当前最后一个日志文件的名称，确定下一个新日志文件的名称，
	// seq （）方法返回当前最后一个日志文件的编号，w.enti记录了当前最后一条日志记录的索引值
	fpath := filepath.Join(w.dir, walName(w.seq()+1, w.enti+1))

	// 从filePipeline获取新建的临时文件，
	newTail, err := w.fp.Open()
	if err != nil {
		return err
	}

	// 将临时文件的句柄保存到WAL.locks中
	w.locks = append(w.locks, newTail)
	prevCrc := w.encoder.crc.Sum32()
	// 创建临时文件对应的encoder实例，并更新到WAL.encoder字段中，
	w.encoder, err = newFileEncoder(w.tail().File, prevCrc)
	if err != nil {
		return err
	}
	// 向临时文件中追加一条crcType类型的日志记录，
	if err = w.saveCrc(prevCrc); err != nil {
		return err
	}
	// 向临时文件中追加一条metadataType类型的日志记录，
	if err = w.encoder.encode(&walpb.Record{Type: metadataType, Data: w.metadata}); err != nil {
		return err
	}
	// 向临时文件中追加一条stateType类型的日志记录，
	if err = w.saveState(&w.state); err != nil {
		return err
	}

	// 通过WAL.sync （）方法将上述修改同步刷新到磁盘上
	if err = w.sync(); err != nil {
		return err
	}
	// 记录当前文件指针的位置，为重命名之后，重新打开文件做准备，
	off, err = w.tail().Seek(0, io.SeekCurrent)
	if err != nil {
		return err
	}
	// 将临时文件重命名成之前得到的新日志文件名称，
	if err = os.Rename(newTail.Name(), fpath); err != nil {
		return err
	}
	start := time.Now()
	// 将重命名这一操作同步刷新到磁盘上，fsync操作不仅会将文件数据刷新到磁盘上，
	// 还会将文件的元数据也刷新到磁盘上（例如，文件的长度和名称等），
	if err = fileutil.Fsync(w.dirFile); err != nil {
		return err
	}
	walFsyncSec.Observe(time.Since(start).Seconds())

	// 关闭临时文件对应的句柄
	newTail.Close()
	// 打开重命名后的新日志文件，
	if newTail, err = fileutil.LockFile(fpath, os.O_WRONLY, fileutil.PrivateFileMode); err != nil {
		return err
	}
	// 将文件指针的位置移动到之前保存的位置
	if _, err = newTail.Seek(off, io.SeekStart); err != nil {
		return err
	}
	// 将WAL.locks中最后一项更新成新日志文件
	w.locks[len(w.locks)-1] = newTail

	prevCrc = w.encoder.crc.Sum32()
	// 创建新日志文件对应的encoder实例，并更新到WAL.encoder字段中，
	w.encoder, err = newFileEncoder(w.tail().File, prevCrc)
	if err != nil {
		return err
	}

	w.lg.Info("created a new WAL segment", zap.String("path", fpath))
	return nil
}

// sync（）方法将日志同步刷新到磁盘上
func (w *WAL) sync() error {
	if w.encoder != nil {
		// 先使用encoder.flush（）方法进行同步刷新，
		if err := w.encoder.flush(); err != nil {
			return err
		}
	}

	if w.unsafeNoSync {
		return nil
	}

	start := time.Now()
	// 使用操作系统的fdatasync将数据真正刷新到磁盘上
	err := fileutil.Fdatasync(w.tail().File)
	// 这里会对该刷新操作的执行时间进行监控，如采刷新操作执行的时间长于指定的时间（默认值是1s),则输出警告日志
	took := time.Since(start)
	if took > warnSyncDuration {
		w.lg.Warn(
			"slow fdatasync",
			zap.Duration("took", took),
			zap.Duration("expected-duration", warnSyncDuration),
		)
	}
	walFsyncSec.Observe(took.Seconds())

	return err
}

func (w *WAL) Sync() error {
	return w.sync()
}

// WAL 日志的文件名中包含了该文件中第一条 E向记录的索引值， WAL. locks 宇段中
// 记录了当前 WAL 实例正在使用的 WAL 文件句柄。在 WAL.ReleaseLockTo （）方法中， 会根据 WAL
// 日志的文件名和快照的元数据，将比较旧的 WAL 日志文件句柄从 WAL. locks 中清除，
func (w *WAL) ReleaseLockTo(index uint64) error {
	w.mu.Lock()
	defer w.mu.Unlock()

	if len(w.locks) == 0 {
		return nil
	}

	var smaller int
	found := false
	for i, l := range w.locks {
		// 遍历 locks 字段， 解析对应的 WAL 日志文件名，获取其中第一条记录的索引值
		_, lockIndex, err := parseWALName(filepath.Base(l.Name()))
		if err != nil {
			return err
		}
		if lockIndex >= index {	// 检测是否可以清除该 WAL 日志文件的句柄
			smaller = i - 1	// 真正要释放的是 smaller 之前的 WAL 日志文件
			found = true
			break
		}
	}

	// 如果没有锁索引大于释放索引，我们可以
	// 释放锁直到最后一个（不包括）。
	if !found {
		smaller = len(w.locks) - 1
	}

	if smaller <= 0 {
		return nil
	}
	// 对 smaller 进行边界检查 ， 省略该检查的相关代码
	for i := 0; i < smaller; i++ {
		if w.locks[i] == nil {
			continue
		}
		w.locks[i].Close()
	}
	w.locks = w.locks[smaller:]	// 清理 smaller 之前的文件句柄

	return nil
}

// 关闭当前WAL文件和目录。
func (w *WAL) Close() error {
	w.mu.Lock()
	defer w.mu.Unlock()

	if w.fp != nil {
		w.fp.Close()
		w.fp = nil
	}

	if w.tail() != nil {
		if err := w.sync(); err != nil {
			return err
		}
	}
	for _, l := range w.locks {
		if l == nil {
			continue
		}
		if err := l.Close(); err != nil {
			w.lg.Error("failed to close WAL", zap.Error(err))
		}
	}

	return w.dirFile.Close()
}

// 追加日志记录的操作
func (w *WAL) saveEntry(e *raftpb.Entry) error {
	// 将Entry记录序列化
	b := pbutil.MustMarshal(e)
	// 将序列化后的数据封装成entryType类型的Record记录
	rec := &walpb.Record{Type: entryType, Data: b}
	// 通过encoder.encode（）方法追加日志记录，
	if err := w.encoder.encode(rec); err != nil {
		return err
	}
	// 更新WAL.enti字段，其中保存了最后一条Entry记录的索引位
	w.enti = e.Index
	return nil
}

func (w *WAL) saveState(s *raftpb.HardState) error {
	if raft.IsEmptyHardState(*s) {
		return nil
	}
	w.state = *s
	b := pbutil.MustMarshal(s)
	rec := &walpb.Record{Type: stateType, Data: b}
	return w.encoder.encode(rec)
}

// WAL.Save（）方法先将待写入的Entry记录封装成entryType类型的Record实例，
// 然后将其序列化并追加到日志段文件中，之后将HardState封装成stateType类型的Record实例，
// 并序列化写入日志段文件中，最后将这些日志记录同步刷新到磁盘。
func (w *WAL) Save(st raftpb.HardState, ents []raftpb.Entry) error {
	w.mu.Lock()	// 加锁同步
	defer w.mu.Unlock()	// 日志记录追加完成之后，释放锁

	// 边界检查，如果待写入的HardSta怡和Entry数量且都为空，则直接返回,否则就需要将修改同步到磁盘上
	if raft.IsEmptyHardState(st) && len(ents) == 0 {
		return nil
	}
	mustSync := raft.MustSync(st, w.state, len(ents))

	// 遍历待写入的Entry数组，将每个Entry实例序列化并封装entryType类型的日志记录，写入日志文件
	for i := range ents {
		if err := w.saveEntry(&ents[i]); err != nil {	// 如果发生异常，则将异常返回
			return err
		}
	}
	// 将状态信息（HardState）序列化并封装成stateType类型的日志记录，写入日志文件
	if err := w.saveState(&st); err != nil {
		return err
	}
	// 获取当前日志段文件的文件指针的位置，
	curOff, err := w.tail().Seek(0, io.SeekCurrent)
	if err != nil {
		return err
	}
	// 如果未写满预分画己的空间，将新日志刷新到磁盘后，即可返回
	if curOff < SegmentSizeBytes {
		if mustSync {
			return w.sync()	// 将上述追加的日志记录同步刷新到磁盘上
		}
		return nil
	}

	return w.cut()	// 当前文件大小已超出了预分配的空间，则需妥进行日志文件的切换
}

func (w *WAL) SaveSnapshot(e walpb.Snapshot) error {
	if err := walpb.ValidateSnapshotForWrite(&e); err != nil {
		return err
	}

	b := pbutil.MustMarshal(&e)

	w.mu.Lock()
	defer w.mu.Unlock()

	rec := &walpb.Record{Type: snapshotType, Data: b}
	if err := w.encoder.encode(rec); err != nil {
		return err
	}
	// 仅当快照位于最后一个索引之前时更新enti 
	if w.enti < e.Index {
		w.enti = e.Index
	}
	return w.sync()
}

func (w *WAL) saveCrc(prevCrc uint32) error {
	return w.encoder.encode(&walpb.Record{Type: crcType, Crc: prevCrc})
}

func (w *WAL) tail() *fileutil.LockedFile {
	if len(w.locks) > 0 {
		return w.locks[len(w.locks)-1]
	}
	return nil
}

func (w *WAL) seq() uint64 {
	t := w.tail()
	if t == nil {
		return 0
	}
	seq, _, err := parseWALName(filepath.Base(t.Name()))
	if err != nil {
		w.lg.Fatal("failed to parse WAL name", zap.String("name", t.Name()), zap.Error(err))
	}
	return seq
}

func closeAll(lg *zap.Logger, rcs ...io.ReadCloser) error {
	stringArr := make([]string, 0)
	for _, f := range rcs {
		if err := f.Close(); err != nil {
			lg.Warn("failed to close: ", zap.Error(err))
			stringArr = append(stringArr, err.Error())
		}
	}
	if len(stringArr) == 0 {
		return nil
	}
	return errors.New(strings.Join(stringArr, ", "))
}
